package com.hp.consumer;

import com.alibaba.cloud.stream.binder.rocketmq.constant.RocketMQConst;
import com.alibaba.cloud.stream.binder.rocketmq.support.RocketMQMessageConverterSupport;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;

import java.util.function.Consumer;

@SpringBootApplication
@Slf4j
//@EnableBinding({PropertiesInSource.class})
public class SpringRocketmqConsumerApplication {

    public static void main(String[] args) {
        SpringApplication.run(SpringRocketmqConsumerApplication.class, args);
    }
    
    /**
     * 使用@StreamListener("input1")时
     * 获取rocketmq控制台上的信息，按照下面的方式获取
     * header 里 key 是 rocketmq_KEYS 和 rocketmq_TAGS
     * 参数可以用 Message 或 @Header 注解拿
     * 
     */
    //@StreamListener(value = "input1", condition = "headers['ROCKET_KEYS'] == '9key'||headers['ROCKET_TAGS'] == '9tag'")
    //public void receiveInput1(String receiveMsg, Message message) {
    //    String keys = String.valueOf(message.getHeaders().get(Constant.ROCKETMQ_KEYS));
    //    String tags = String.valueOf(message.getHeaders().get(Constant.ROCKETMQ_TAGS));
    //    String msgId = String.valueOf(message.getHeaders().get(Constant.ROCKETMQ_MESSAGE_ID));
    //    System.out.println("input1 receive: " + receiveMsg);
    //    System.out.println("input1 keys: " + keys);
    //    System.out.println("input1 tags: " + tags);
    //    System.out.println("input1 msgId: " + msgId);
    //}

//    @StreamListener("input2")
//    public void receiveInput2(String receiveMsg) {
//        System.out.println("input2 receive: " + receiveMsg);
//    }

//    @StreamListener("input3")
//    public void receiveInput3(@Payload Foo foo) {
//        System.out.println("input3 receive: " + foo);
//    }

//    @StreamListener("input4")
//    public void receiveTransactionalMsg(String transactionMsg) {
//        System.out.println("input4 receive transaction msg: " + transactionMsg);
//    }

//    @StreamListener("input6")
//    public void receiveInput6(String receiveMsg) {
//        System.out.println("input6 receive: " + receiveMsg);
//    }


    @Bean
    public Consumer<Message<User>> inputRocket1() {
        return str -> {
            // 收到消息在这里做一些处理
            log.info("inputRocket1 message: {}", str);
            //log.info("inputRocket1 message: tag-> {}", str.getHeaders().get(Constant.ROCKETMQ_TAGS));
            log.info("inputRocket1 message: tag-> {}", str.getHeaders().get(RocketMQMessageConverterSupport.toRocketHeaderKey(RocketMQConst.Headers.TAGS)));
            log.info("inputRocket1 message: key-> {}", str.getHeaders().get(RocketMQMessageConverterSupport.toRocketHeaderKey(RocketMQConst.Headers.KEYS)));
        };
    }

    @Bean
    public Consumer<Message<User>> inputRocket2() {
        return str -> {
            log.info("inputRocket2 message: {}", str.getPayload());
        };
    }


//    // 第一种方式(官方推荐)
//    @Bean
//    public Function<Flux<Message<User>>, Mono<Void>> inputRocket1() {
//        return flux -> flux.map(str -> {
//            // 收到消息在这里做一些处理
//            log.info("inputRocket1 message: {}", str);
//            log.info("inputRocket1 message: tag-> {}", str.getHeaders().get(Constant.ROCKETMQ_TAGS));
//            log.info("inputRocket1 message: key-> {}", str.getHeaders().get(Constant.ROCKETMQ_KEYS));
//            return str;
//        }).then();
//    }
//
//    // 第二种方式
//// 注意使用Flux 要调用 subscribe 不然这个方法不会被消费
//    @Bean
//    public Consumer<Flux<Message<User>>> inputRocket2() {
//        return flux -> flux.map(message -> {
//            log.info("inputRocket2 message: {}", message.getPayload());
//            return message;
//        }).subscribe();
//    }
//    // 或
//    @Bean
//    public Consumer<Message<String>> consumerEvent() {
//        return message -> System.out.println(message.getPayload());
//    }



}
